package com.flinkjar.flinkjardemo;

import com.flinkjar.flinkjardemo.bean.UserProduct;
import com.flinkjar.flinkjardemo.common.CustomBatchSinkFunction;
import com.flinkjar.flinkjardemo.common.CustomBatchSinkFunctionV2;
import com.flinkjar.flinkjardemo.common.CustomDeserializationSchema;
import com.flinkjar.flinkjardemo.common.CustomSinkFunction;
import com.flinkjar.flinkjardemo.config.DBConfig;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;
import java.util.Properties;

public class MysqlMain {


    public static void main(String[] args) throws Exception {
        Properties properties = new Properties();
        properties.put("jdbc.properties.useSSL", "false");


        MySqlSource<UserProduct> mySqlSource = MySqlSource.<UserProduct>builder()
                .hostname(DBConfig.hostname_source)
                .port(DBConfig.port_source)
                .username(DBConfig.username_source)
                .password(DBConfig.password_source)
                // 设置捕获的数据库
                .databaseList(DBConfig.database_source)
                // 设置捕获表
                .tableList(String.join(".", DBConfig.database_source, DBConfig.table_source))
                // 数据库连接池大小。
                .connectionPoolSize(5)
                .connectTimeout(Duration.ofSeconds(3L))
                .jdbcProperties(properties)
                // 全量+增量
                .startupOptions(StartupOptions.initial())
                // 在 Flink 的 CDC（Change Data Capture）连接器中，fetchSize 参数用于指定每次从数据库中读取的记录数目。
                /**
                 * 注意:fetchSize大小 + source并行度 + sink并行度 需要很大的内存资源
                 * eg:
                 * fetchSize = 500, source并行度=15, sink并行度=20 大约需要 210GB 的内存资源
                 * fetchSize = 500, source并行度=5, sink并行度=25 大约需要 260GB 的内存资源
                 * fetchSize = 1000, source并行度=5, sink并行度=30 大约需要 310GB 的内存资源
                 */
                .fetchSize(1000)
                // 反序列化器，将SourceRecord类型记录反序列化到指定类型
                .deserializer(new CustomDeserializationSchema())
                // 时区
                .serverTimeZone("Asia/Shanghai")
                // server id 需要全局唯一，server id 是一个int类型整数。 在 CDC 2.x 版本中，source 的每个并发都需要一个server id，建议合理规划好server id，比如作业的 source 设置成了四个并发，可以配置 'server-id' = '5001-5004', 这样每个 source task 就不会冲突了。
                .serverId("7000-7010")
                .build();


        // 创建 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 开启检查点, 每3秒进行一次 Checkpoint
        env.enableCheckpointing(3000L);

        /**
         * 在Flink1.12以前，旧的添加source的方式，是调用执行环境的addSource()方法：
         * DataStream<String> stream = env.addSource(...);
         *
         * 方法传入的参数是一个“源函数”（source function），需要实现SourceFunction接口。
         * 从Flink1.12开始，主要使用流批统一的新Source架构：
         * DataStreamSource<String> stream = env.fromSource(…)
         */
        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "guess-mysql")
                // 设置 2 个并行源任务
                .setParallelism(1)
                // 根据hash sink到不同的表(增量阶段使用)
                //.addSink(new CustomSinkFunction(DBConfig.tableCount))
                // 批处理100(全量阶段使用)
                .addSink(new CustomBatchSinkFunctionV2(10, DBConfig.tableCount))
                // 此接收器的名称。此名称在运行时期间由可视化和日志记录使用。
                .name("guess-mysql")
                // 对sink使用 n 个并行来保持消息顺序, 增量阶段sink并行度推荐=1
                .setParallelism(1);

        env.execute("flink-jar作业");

    }


}
